package com.woniuxy.litener;


import com.woniuxy.config.RabbitConfig;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.woniuxy.pojo.SysLogOpt;
import com.rabbitmq.client.Channel;

import com.woniuxy.service.SysLogOptService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.stereotype.Component;
import org.springframework.amqp.core.Message;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

@Component
@Slf4j
public class FlushToDBListener {

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    private static final String LOG_LIST_KEY = "log:list";
    @Autowired
    private SysLogOptService sysLogOptService;
    @Autowired
    private ObjectMapper objectMapper;


    @RabbitListener(queues = RabbitConfig.QUEUE_DB)
    public void flushCacheToDB(List<Object> logs, Channel channel,
                               @Header(AmqpHeaders.DELIVERY_TAG) Long tag,
                               @Headers Map<String, Object> headers,
                               Message message) throws IOException {

        try {
            List<SysLogOpt> logOpts = logs.stream()
                    .map(obj -> objectMapper.convertValue(obj, SysLogOpt.class))
                    .collect(Collectors.toList());
//            System.out.println(1 / 0);//手动异常

//        即使数据大于200，也会别分批次入库，不会丢失
            sysLogOptService.saveBatch(logOpts, 200);//这个方法默认保存一千条,

            channel.basicAck(tag, false); // 正常确认
        } catch (Exception e) {

            log.error("日志批量入库失败，准备发送到死信队列", e);
            Boolean redelivered = message.getMessageProperties().getRedelivered();

            //获取重投标记
            if (redelivered) {
                log.error("重试消费消息重试超过次数，拒绝消息");
                channel.basicReject(tag, false);
            } else {
                log.error("重试消费消息重试一次");
                channel.basicNack(tag, false, true);
            }
        }

    }
}
